package com.buddy.java.thread.common;

import com.buddy.java.thread.common.AlarmAgent;
import com.buddy.java.thread.common.AlarmInfo;
import com.buddy.java.thread.common.AlarmType;
import com.buddy.java.thread.tow_phase_termination.AbstractTerminatableThread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

public class AlarmSendingThread extends AbstractTerminatableThread {

    private final AlarmAgent alarmAgent = new AlarmAgent();

    //告警队列
    private final BlockingQueue<AlarmInfo> alarmQueue;
    private final ConcurrentMap<String, AtomicInteger> submittedAlarmRegistry;

    public AlarmSendingThread() {
        alarmQueue = new ArrayBlockingQueue<AlarmInfo>(100);
        submittedAlarmRegistry = new ConcurrentHashMap<String, AtomicInteger>();
        alarmAgent.init();
    }

    protected void doRun() throws Exception {
        AlarmInfo alarmInfo;
        alarmInfo = alarmQueue.take();
        terminationToken.reservations.decrementAndGet();

        //将告警信息发送到告警服务器
        alarmAgent.sendAlarm(alarmInfo);

        //处理恢复告警,将相应的故障告警从注册表中删除,使得相应故障恢复后,若再次出现相同故障,该故障信息能够上报到服务器
        if (AlarmType.RESUME == alarmInfo.alarmType) {
            String key = AlarmType.FAULT.toString()+":"+alarmInfo.getId()+"@"+alarmInfo.getExtraInfo();
            submittedAlarmRegistry.remove(key);

            key = AlarmType.RESUME.toString()+":"+alarmInfo.getId()+"@"+alarmInfo.getExtraInfo();
            submittedAlarmRegistry.remove(key);
        }
    }

    public int sendAlarm(final AlarmInfo alarmInfo) {
        AlarmType alarmType = alarmInfo.alarmType;
        String id = alarmInfo.getId();
        String extraInfo = alarmInfo.getExtraInfo();

        if (terminationToken.isToShutdown()) {
            //记录告警
            System.err.println("rejected alarm: " + id + "," + extraInfo);
            return -1;
        }

        int duplicateSubmissionCount = 0;
        try {
            AtomicInteger prevSubmittedCounter;
            prevSubmittedCounter = submittedAlarmRegistry.putIfAbsent(alarmType.toString() + ":" + id + "@" + extraInfo, new AtomicInteger(0));
            if (null == prevSubmittedCounter) {
                terminationToken.reservations.incrementAndGet();
                alarmQueue.put(alarmInfo);
            } else {
                //故障未修复,不用重复发送告警信息给服务器,仅增加计数
                duplicateSubmissionCount = prevSubmittedCounter.incrementAndGet();
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return duplicateSubmissionCount;
    }


    protected void doCleanup(Exception exp){
        if (null != exp && !(exp instanceof InterruptedException)){
            exp.printStackTrace();
        }
        alarmAgent.disconnect();
    }
}
